#include "voda.h"
#include "application.h"
#include "thread_pool.h"

namespace voda
{
	boost::shared_ptr<thread_pool> thread_pool::create (
			boost::function<void(thread_pool*)> const & deleter
		)
	{
		if (deleter)
		{
			return boost::shared_ptr<thread_pool> (new thread_pool, deleter);
		}
		else
		{
			return boost::shared_ptr<thread_pool> (new thread_pool);
		}
	}

	thread_pool::thread_pool (void)
		: io_service_ ()
		, strand_ (io_service_)
	{
	}

	thread_pool::~thread_pool (void)
	{
	}

	boost::shared_ptr<iapplication> thread_pool::create_application ()
	{
		return boost::shared_ptr<iapplication> (
				application::create (
					shared_from_this (), 
						boost::bind(
							&thread_pool::application_deleter, 
							shared_from_this (),
							_1
						)
					)
				);
	}

	std::size_t thread_pool::run (
			size_t const threads
		)
	{
		boost::atomic<std::size_t> total_count (0);
		static const std::size_t default_threads = boost::thread::hardware_concurrency () * 2;
		std::size_t threads_count (threads == 0 ? default_threads : std::min<std::size_t> (threads, 1024));
		boost::thread_group thread_group;

		for (std::size_t i = 0; i<threads; ++i)
		{
			thread_group.create_thread (
				boost::bind (
					&thread_pool::worker_thread, 
					shared_from_this (), 
					boost::ref (io_service_), 
					boost::ref (total_count)
				)
			);
		}

		worker_thread (io_service_, total_count);
		thread_group.join_all ();
		return total_count;
	}

	void thread_pool::post (
			boost::function<void()> const & handler
		)
	{
		strand_.post (
			boost::bind (
				&thread_pool::call_function, 
				shared_from_this (), 
				handler
			)
		);
	}

	void thread_pool::io_post (
			boost::function<void()> const & handler
		)
	{
		io_service_.post (
			boost::bind (
				&thread_pool::call_function, 
				shared_from_this (), 
				handler
			)
		);
	}

	void thread_pool::stop ()
	{
		io_service_.stop ();
	}

	boost::asio::io_service & thread_pool::get_io_service ()
	{
		return io_service_;
	}
	
	boost::asio::io_service::strand & thread_pool::get_strand ()
	{
		return strand_;
	}

	void thread_pool::application_deleter (
			application *papplication
		)
	{
		delete papplication;
	}

	void thread_pool::worker_thread (
			boost::asio::io_service & io_service, 
			boost::atomic<std::size_t> & total_count
		)
	{
		thread_init init;

		for (;;)
		{
			try
			{
				total_count += io_service_.run ();
				return;
			}
			catch (std::exception &)
			{
				// LOGE ("Io handler failed: %d", e.what());
			}
		}
	}

	void thread_pool::call_function (
			boost::function<void()> const & fn
		)
	{
		if (fn)
		{
			fn ();
		}
	}
}